package org.example.kafka.concurrent;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.metrics.stats.Percentiles;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.example.kafka.common.CommonConstant;
import org.example.kafka.common.KafkaConstants;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.*;

/**
 * 多线程下正确的使用消费者，需要记住，一个线程一个消费者
 */
public class KafkaConConsumer {
    private static   final int THREAD_POOL_SIZE = 8;
    private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
            .setNameFormat("hyn-demo-pool-%d").build();
    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(THREAD_POOL_SIZE,
            THREAD_POOL_SIZE,
            0L,
            TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(10),
            namedThreadFactory,
            new ThreadPoolExecutor.AbortPolicy());

    private static class ConsumerWorker implements Runnable{
        private KafkaConsumer<String,String> consumer;

        public ConsumerWorker(Map<String, Object> config, String topic) {
            Properties properties = new Properties();
            properties.putAll(config);
            this.consumer = new KafkaConsumer<String, String>(properties);
            // 消费者订阅主题
            consumer.subscribe(Collections.singletonList(topic));
        }
        @Override
        public void run() {
            final String id = Thread.currentThread().getId()
                    +"-"+System.identityHashCode(consumer);
            try {
                while(true){
                    ConsumerRecords<String, String> records
                            = consumer.poll(500);
                    for(ConsumerRecord<String, String> record:records){
                        System.out.println(id+"|"+String.format(
                                "主题：%s，分区：%d，偏移量：%d，" +
                                        "key：%s，value：%s",
                                record.topic(),record.partition(),
                                record.offset(),record.key(),record.value()));
                        //do our work
                    }
                }
            } finally {
                consumer.close();
            }
        }
    }

    public static void main(String[] args) {
        //消费配置的实例
        Map<String,Object> config = KafkaConstants.consumerConfigMap("concurrent", StringDeserializer.class,
                StringDeserializer.class);
        for (int i=0;i< CommonConstant.CONCURRENT_PARTITIONS_COUNT;i++){
            executor.submit(new ConsumerWorker(config, CommonConstant.CONCURRENT_USER_INFO_TOPIC));
        }
    }
}
